/*
 * Copyright (c) 2009, Katholieke Universiteit Leuven
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 *     * Redistributions of source code must retain the above copyright notice,
 *       this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright notice,
 *       this list of conditions and the following disclaimer in the documentation
 *       and/or other materials provided with the distribution.
 *     * Neither the name of the Katholieke Universiteit Leuven nor the names of
 *       its contributors may be used to endorse or promote products derived from
 *       this software without specific prior written permission.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
/**
 * @addtogroup events
 * @{
*/
/**
 * @file
 * Implementation of the LooCI Event Bus
 * @author
 * Wouter Horré <wouter.horre@cs.kuleuven.be>
 */
#include "events.h"
#include "events_private.h"
#include "lib/vector.h"
#include "memb.h"
#include "networking_private.h"
#include "components_private.h"

//#ifdef CONTIKI_TARGET_MINIMAL_NET
#include <string.h>
//#endif

#ifdef LOOCI_EVENTS_DEBUG
#include <stdio.h>
#ifdef CONTIKI_TARGET_AVR_RAVEN
#include <avr/pgmspace.h>
#define PRINTF(FORMAT,args...) printf_P(PSTR(FORMAT),##args)
#else
#define PRINTF printf
#endif // CONTIKI_TARGET_AVR_RAVEN
#else
#define PRINTF(...)
#endif

#define MIN(a, b) ((a) < (b)? (a): (b))

#define LOOCI_EVENTS_QUEUE_SIZE 5

/**
 * @internal
 * @name Internal contiki definitions
 * @{
 */
//! @internal
//! @todo how many events do we want to queue?
MEMB(events, struct looci_event, LOOCI_EVENTS_QUEUE_SIZE);
PROCESS(looci_event_manager, "LooCI Event Manager");
/** @} */

// Function prototypes
static void dispatch_event(struct looci_event *);
static void dispatch_remote_event(struct looci_event *);

// Data structures
struct subscription {
  looci_eventtype_t type;
  struct component * source;
  struct component * destination; 
};
struct subscription_to_remote {
  looci_eventtype_t type;
  struct component * source;
  u8_t refcount;
  peer_id_t peer;
};
struct subscription_from_remote {
  looci_eventtype_t type;
  struct component * destination;
  peer_id_t source_node;
  u8_t source_cid;
};

// A list of subscriptions
VECTOR(subscriptions, struct subscription);
VECTOR(subscriptions_to_remote, struct subscription_to_remote);
VECTOR(subscriptions_from_remote, struct subscription_from_remote);

PROCESS_THREAD(looci_event_manager, ev, data)
{
  PROCESS_BEGIN();

  looci_process_event = process_alloc_event();
  looci_remote_event = process_alloc_event();
  looci_delay_event = process_alloc_event();

  memb_init(&events);
  vector_libinit();
  vector_init(&subscriptions);
  vector_init(&subscriptions_to_remote);
  vector_init(&subscriptions_from_remote);

  PRINTF("[LooCI EM] Started\r\n");

  while(1) {
    PROCESS_WAIT_EVENT_UNTIL(ev==looci_process_event || ev==looci_remote_event);
    PRINTF("[LooCI EM] Processing event\r\n");
    if(ev==looci_process_event) {
      dispatch_event((struct looci_event *)data);
      memb_free(&events, data);
    } else if(ev==looci_remote_event) {
      dispatch_remote_event((struct looci_event *)data);
      memb_free(&events, data);
    }
  }

  PROCESS_END();
}

int looci_event_publish(looci_eventtype_t type, void * payload, u8_t len) 
{
  struct looci_event * e = (struct looci_event *) memb_alloc(&events);
  if(e==NULL) {
    // event queue is full, too bad ...
    PRINTF("[LooCI EM] Dropping local event because the queue is full\r\n");
    return LOOCI_ERR_EVENTNOTPUBLISHED;
  }
  e->type = type;
  e->source_node = PEER_ID_NONE; // PEER_ID_NONE indicates a local event
  struct component * c = (struct component *) components_get_by_process(PROCESS_CURRENT());
  e->source_cid = c->id;
  e->len = MIN(len, LOOCI_EVENT_PAYLOAD_MAXLEN);
  memcpy(e->payload, payload, e->len);

  process_post(&looci_event_manager, looci_process_event, e);

  return LOOCI_SUCCESS;
}

void handle_remote_event(struct looci_event * event) {
  struct looci_event * e = (struct looci_event *) memb_alloc(&events);
  if(e == NULL) {
    PRINTF("[LooCI EM] Dropping remote event because queue is full\r\n");
    return;
  }
  memcpy(e, event, sizeof(struct looci_event));

  PRINTF("[LooCI EM] Putting remote event in the queue\r\n");
  process_post(&looci_event_manager, looci_remote_event, e);
}

/**
 * @internal
 * Dispatch an event.
 *
 * @param em The event to dispatch, including meta data
 */
static void dispatch_event(struct looci_event * e) {
  PRINTF("[LooCI EM] Dispatching event\r\n");
    // Check local subscriptions
    struct subscription * sub;
    int i;
    for(i=0; i < vector_len(&subscriptions); ++i){
      sub = (struct subscription *) vector_get(&subscriptions, i);
      PRINTF("[LooCI EM] We have a subscription for event type %u\r\n", sub->type);
      if(looci_eventtype_isa(e->type, sub->type) && (sub->source==SOURCE_ANY || e->source_cid==sub->source->id)) {
        process_post_synch(sub->destination->process, looci_process_event, e);
      }
    }
    // Check remote subscriptions
    struct subscription_to_remote * sub_to;
    for(i=0; i < vector_len(&subscriptions_to_remote); ++i) {
      sub_to = (struct subscription_to_remote *) vector_get(&subscriptions_to_remote, i);
      PRINTF("[LooCI EM] We have a remote subscription for event type %u with refcount %u\r\n", sub_to->type, sub_to->refcount);
      if(looci_eventtype_isa(e->type, sub_to->type) \
          && (sub_to->source == SOURCE_ANY || e->source_cid==sub_to->source->id)) {
        nf_send_event(e, sub_to->peer);
      }
    }
}

/**
 * @internal
 * Dispatch an event from a remote node.
 *
 * @param em The event to dispatch, including meta data
 */
static void dispatch_remote_event(struct looci_event * e) {
  PRINTF("[LooCI EM] Dispatching remote event from peer %u, component %u, eventtype: %u, length: %u\r\n", \
      e->source_node, e->source_cid, e->type, e->len);
  // Check subscriptions for incoming events
  struct subscription_from_remote * sub;
  int i;
  for(i=0; i < vector_len(&subscriptions_from_remote); ++i) {
    sub = (struct subscription_from_remote *) vector_get(&subscriptions_from_remote, i);
    PRINTF("[LooCI EM] We have a subscription for incoming event type %u\r\n", sub->type);
    if(looci_eventtype_isa(e->type, sub->type) \
        && (sub->source_node==PEER_ID_NONE || sub->source_node==e->source_node) \
        && (sub->source_cid==COMPONENT_ID_NONE || e->source_cid==sub->source_cid)) {
      process_post_synch(sub->destination->process, looci_process_event, e);
    }
  }
}

/**
 * @internal
 */
void events_add_local_subscription(looci_eventtype_t type, struct component * source, struct component * destination)
{
  PRINTF("[LooCI EM] Adding subscription for event type %u: source '%s', destination '%s'\r\n", type, \
      (source!=SOURCE_ANY ? COMPONENT_TYPE(source): "ANY"), COMPONENT_TYPE(destination));
  // @TODO should we check for duplicates?
  struct subscription sub;
  sub.type = type;
  sub.source = source;
  sub.destination = destination;
  vector_add(&subscriptions, &sub);
}

/**
 * @internal
 */
void events_add_remote_subscription_to(looci_eventtype_t type, struct component * source, peer_id_t peer) 
{
  PRINTF("[LooCI EM] Adding remote subscription for event type %u: source '%s', destination %u\r\n", type, \
      (source!=SOURCE_ANY ? COMPONENT_TYPE(source): "ANY"), peer);
  // Check whether such subscription is already present and change refcount if needed
  // @TODO this makes the operation not idempotent -> might cause troubles with reliability?
  int i;
  struct subscription_to_remote * psub;
  for(i=0; i < vector_len(&subscriptions_to_remote); ++i) {
    psub = (struct subscription_to_remote *) vector_get(&subscriptions_to_remote, i);
    if(psub->type == type && psub->source == source && psub->peer == peer) {
      psub->refcount++;
      return;
    }
  }
  // Not found, add a new one
  struct subscription_to_remote sub;
  sub.type = type;
  sub.source = source;
  sub.peer = peer;
  sub.refcount = 1;
  vector_add(&subscriptions_to_remote, &sub);
}

/**
 * @internal
 */
void events_add_remote_subscription_from(looci_eventtype_t type, peer_id_t source_node, u8_t source_cid, struct component * destination)
{
  PRINTF("[LooCI EM] Adding subscription for incoming remote event type %u: source '%u/%u', destination %s\r\n", type, \
      source_node, source_cid, COMPONENT_TYPE(destination));
  // @TODO should we check for duplicates?
  struct subscription_from_remote sub;
  sub.type = type;
  sub.source_node = source_node;
  sub.source_cid = source_cid;
  sub.destination = destination;
  vector_add(&subscriptions_from_remote, &sub);
}

/**
 * @internal
 */
void events_remove_local_subscription(looci_eventtype_t type, struct component * source, struct component * destination)
{
  PRINTF("[LooCI EM] Removing subscription for event type %u: source '%s', destination '%s'\r\n", type, \
      (source!=SOURCE_ANY ? COMPONENT_TYPE(source): "ANY"), COMPONENT_TYPE(destination));
  int i;
  struct subscription * sub;
  for(i=0; i < vector_len(&subscriptions); ++i) {
    sub = (struct subscription *) vector_get(&subscriptions, i);
    if(sub->type == type && sub->source == source && sub->destination == destination) {
      vector_remove(&subscriptions, i);
      return;
    }
  }
}

/**
 * @internal
 */
void events_remove_remote_subscription_to(looci_eventtype_t type, struct component * source, peer_id_t peer)
{
  PRINTF("[LooCI EM] Removing remote subscription for event type %u: source '%s', destination %u\r\n", type, \
      (source!=SOURCE_ANY ? COMPONENT_TYPE(source): "ANY"), peer);
  // Check whether such subscription is present and decrease refcount
  // Remove if refcount reaches 0
  int i;
  struct subscription_to_remote * psub;
  for(i=0; i < vector_len(&subscriptions_to_remote); ++i) {
    psub = (struct subscription_to_remote *) vector_get(&subscriptions_to_remote, i);
    if(psub->type == type && psub->source == source && psub->peer == peer) {
      psub->refcount--;
      if(psub->refcount == 0) {
        // not needed anymore
        vector_remove(&subscriptions_to_remote, i);
      }
      return;
    }
  }
}

/**
 * @internal
 */
void events_remove_remote_subscription_from(looci_eventtype_t type, peer_id_t source_node, u8_t source_cid, struct component * destination)
{
  PRINTF("[LooCI EM] Removing subscription for incoming remote event type %u: source '%u/%u', destination %s\r\n", type, \
      source_node, source_cid, COMPONENT_TYPE(destination));
  int i;
  struct subscription_from_remote * fsub;
  for(i=0; i < vector_len(&subscriptions_from_remote); ++i) {
    fsub = (struct subscription_from_remote *) vector_get(&subscriptions_from_remote, i);
    if(fsub->type == type && fsub->source_node == source_node && fsub->source_cid == source_cid
        && fsub->destination == destination) {
      vector_remove(&subscriptions_from_remote, i);
      return;
    }
  }
}

/**
 * @internal
 */
void events_unwire_component(u8_t cid) {
  int i;
  // check local subscriptions
  struct subscription * lsub;
  for(i=0; i < vector_len(&subscriptions); ++i) {
    lsub = (struct subscription *) vector_get(&subscriptions, i);
    if(lsub->source->id == cid || lsub->destination->id == cid) {
      vector_remove(&subscriptions, i);
      i--; // vector_remove shifts all entries to the front!
    }
  }
  // check remote incoming subscriptions
  struct subscription_from_remote * fsub;
  for(i=0; i < vector_len(&subscriptions_from_remote); ++i) {
    fsub = (struct subscription_from_remote *) vector_get(&subscriptions_from_remote, i);
    if(fsub->destination->id == cid) {
      vector_remove(&subscriptions_from_remote, i);
      i--; // vector_remove shifts all entries to the front!
    }
  }
  // check remote outgoing subscriptions
  struct subscription_to_remote * tsub;
  for(i=0; i < vector_len(&subscriptions_to_remote); ++i) {
    tsub = (struct subscription_to_remote *) vector_get(&subscriptions_to_remote, i);
    if(tsub->source->id == cid) {
      vector_remove(&subscriptions_to_remote, i);
      i--; // vector_remove shifts all entries to the front!
    }
  }
}

/**
 * @internal
 */
bool events_getlocalwires(looci_eventtype_t eventtype, u8_t cid, u8_t * buffer, u8_t * size) {
  u8_t i;
  u8_t nb;
  struct subscription * lsub;
  for(i=0, nb=0; nb < *size && i < vector_len(&subscriptions); ++i) {
    lsub = (struct subscription *) vector_get(&subscriptions, i);
    if(lsub->source->id == cid && lsub->type == eventtype) {
      buffer[nb++] = lsub->destination->id;
    }
  }
  *size = nb;
  if(i == vector_len(&subscriptions)) { // we got them all
    return true;
  } else {
    return false;
  }
}

/**
 * @internal
 */
bool events_getremotewires_to(looci_eventtype_t eventtype, u8_t cid, peer_id_t * buffer, u8_t * size) {
  u8_t i;
  u8_t nb;
  struct subscription_to_remote * tsub;
  for(i=0, nb=0; nb < *size && i < vector_len(&subscriptions_to_remote); ++i) {
    tsub = (struct subscription_to_remote *) vector_get(&subscriptions_to_remote, i);
    if(tsub->source->id == cid && tsub->type == eventtype) {
      buffer[nb++] = tsub->peer;
    }
  }
  *size = nb;
  if(i==vector_len(&subscriptions_to_remote)) { // we got them all
    return true;
  } else {
    return false;
  }
}

/**
 * @internal
 */
bool events_getremotewires_from(looci_eventtype_t eventtype, u8_t cid, struct wire_from * buffer, u8_t * size){
  u8_t i;
  u8_t nb;
  struct subscription_from_remote * fsub;
  for(i=0, nb=0; nb < *size && i < vector_len(&subscriptions_from_remote); ++i) {
    fsub = (struct subscription_from_remote *) vector_get(&subscriptions_from_remote, i);
    if(fsub->type == eventtype && fsub->destination->id == cid) {
      buffer[nb].source_node = fsub->source_node;
      buffer[nb].source_cid = fsub->source_cid;
      nb++;
    }
  }
  PRINTF("[LooCI Events] Found %u incoming wires\r\n", nb);
  *size = nb;
  if(i==vector_len(&subscriptions_from_remote)) { // we got them all
    PRINTF("[LooCI Events] We found all incoming wires\r\n");
    return true;
  } else {
    PRINTF("[LooCI Events] We didn't find all incoming wires\r\n");
    return false;
  }
}  
/** @} */
